﻿using Microsoft.Extensions.ObjectPool;
using System.Buffers;
using System.IO.Pipelines;
using System.Net;
using System.Net.Sockets;
using System.Text;

namespace PipelineTcpServer
{
	public partial class MainForm : Form
	{
		private CancellationTokenSource _cts;
		private TcpServer _server;
		private int _clientCount;

		public MainForm()
		{
			InitializeComponent();
			btnStart.Enabled = true;
			btnStop.Enabled = false;
		}

		private async void btnStart_Click(object sender, EventArgs e)
		{
			if (!int.TryParse(txtPort.Text, out int port) || port < 1 || port > 65535)
			{
				MessageBox.Show("请输入有效的端口号 (1-65535)");
				return;
			}

			btnStart.Enabled = false;
			btnStop.Enabled = true;
			lstClients.Items.Clear();
			_clientCount = 0;

			_cts = new CancellationTokenSource();
			_server = new TcpServer();

			// 设置事件处理
			_server.ClientConnected += OnClientConnected;
			_server.ClientDisconnected += OnClientDisconnected;
			_server.MessageReceived += OnMessageReceived;
			_server.ServerLog += OnServerLog;

			try
			{
				await _server.StartAsync(port, _cts.Token);
				LogMessage($"服务器已在端口 {port} 上启动");
			}
			catch (Exception ex)
			{
				LogMessage($"服务器启动失败: {ex.Message}");
				btnStart.Enabled = true;
				btnStop.Enabled = false;
			}
		}

		private async void btnStop_Click(object sender, EventArgs e)
		{
			btnStop.Enabled = false;
			_cts?.Cancel();

			if (_server != null)
			{
				await _server.StopAsync();
				_server = null;
			}

			btnStart.Enabled = true;
			LogMessage("服务器已停止");
		}

		private void OnClientConnected(object sender, ClientInfo client)
		{
			Invoke(new Action(() =>
			{
				_clientCount++;
				lstClients.Items.Add($"{client.Id}");
				lblStatus.Text = $"已连接客户端: {_clientCount}";
			}));
		}

		private void OnClientDisconnected(object sender, ClientInfo client)
		{
			Invoke(new Action(() =>
			{
				_clientCount--;
				for (int i = 0; i < lstClients.Items.Count; i++)
				{
					if (lstClients.Items[i].ToString().Contains(client.Id))
					{
						lstClients.Items.RemoveAt(i);
						break;
					}
				}
				lblStatus.Text = $"已连接客户端: {_clientCount}";
			}));
		}

		private void OnMessageReceived(object sender, ClientMessage message)
		{
			Invoke(new Action(() =>
			{
				var item = new ListViewItem(new[] {
					DateTime.Now.ToString("HH:mm:ss"),
					message.Client.Id.Substring(0, 8),
					message.Message
				});
				lstMessages.Items.Add(item);
				item.EnsureVisible();
			}));
		}

		private void OnServerLog(object sender, string message)
		{
			LogMessage(message);
		}

		private void LogMessage(string message)
		{
			Invoke(new Action(() =>
			{
				txtLog.AppendText($"[{DateTime.Now:HH:mm:ss}] {message}{Environment.NewLine}");
			}));
		}

		private void MainForm_FormClosing(object sender, FormClosingEventArgs e)
		{
			_cts?.Cancel();
			_server?.StopAsync().Wait();
		}

		private void btnSend_Click(object sender, EventArgs e)
		{
			//lstClients.SelectedItem
            _server.Send(lstClients.SelectedItem as string, "Hello World");
		}
	}

	public class ClientInfo
    {
        public string Id { get; } = Guid.NewGuid().ToString();
        public EndPoint RemoteEndPoint { get; set; }
		public TcpClient tcpClient { get; set; }

	}

    public class ClientMessage
    {
        public ClientInfo Client { get; set; }
        public string Message { get; set; }
    }

    public class TcpServer
    {
        private TcpListener _listener;
        private CancellationTokenSource _cts;
        private readonly object _syncRoot = new object();
        private int _connectionCount;

        private DefaultPooledObjectPolicy<ClientInfo> policy;
        private DefaultObjectPool<ClientInfo> pool;
        private Extra<ClientInfo> extra;

		public event EventHandler<ClientInfo> ClientConnected;
        public event EventHandler<ClientInfo> ClientDisconnected;
        public event EventHandler<ClientMessage> MessageReceived;
        public event EventHandler<string> ServerLog;

        public async Task StartAsync(int port, CancellationToken cancellationToken)
        {
			policy = new DefaultPooledObjectPolicy<ClientInfo>();
			pool = new DefaultObjectPool<ClientInfo>(policy);
			extra = new Extra<ClientInfo>();

			_cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            _listener = new TcpListener(IPAddress.Any, port);
            _listener.Start();

            LogMessage($"服务器开始监听端口 {port}...");

            try
            {
                while (!_cts.IsCancellationRequested)
                {
                    var client = await _listener.AcceptTcpClientAsync();
					_ = HandleClientAsync(client);
				}
            }
            catch (ObjectDisposedException)
            {
                // 正常关闭
            }
            catch (Exception ex)
            {
                LogMessage($"接受客户端连接时出错: {ex.Message}");
            }
        }

        public async Task StopAsync()
        {
            _cts?.Cancel();
            _listener?.Stop();
            await Task.CompletedTask;
        }

        public void Send(string key , string message)
        {
			var clientInfo = extra.Get(key);
			byte[] data = System.Text.Encoding.UTF8.GetBytes(message);
			clientInfo.tcpClient.Client.Send(data);
		}

        private async Task HandleClientAsync(TcpClient tcpClient)
        {
			var clientInfo = pool.Get();
			//var clientInfo = new ClientInfo
			//         {
			//             RemoteEndPoint = tcpClient.Client.RemoteEndPoint
			//         };

			clientInfo.RemoteEndPoint = tcpClient.Client.RemoteEndPoint;
			clientInfo.tcpClient = tcpClient;
			extra.Set(clientInfo.Id,clientInfo);
			Interlocked.Increment(ref _connectionCount);
            LogMessage($"客户端连接: {clientInfo.RemoteEndPoint} [连接数: {_connectionCount}]");
            ClientConnected?.Invoke(this, clientInfo);

			try
            {
                using (tcpClient)
                using (var stream = tcpClient.GetStream())
                {
                    var pipe = new Pipe();
                    var writing = FillPipeAsync(stream, pipe.Writer, _cts.Token);
                    var reading = ReadPipeAsync(pipe.Reader, clientInfo, _cts.Token);

                    await Task.WhenAll(writing, reading);
                }
            }
            catch (Exception ex)
            {
                LogMessage($"处理客户端错误 ({clientInfo.Id}): {ex.Message}");
            }
            finally
            {
                Interlocked.Decrement(ref _connectionCount);
				extra.Remove(clientInfo.Id);
				pool.Return(clientInfo);
				LogMessage($"客户端断开: {clientInfo.RemoteEndPoint} [连接数: {_connectionCount}]");
                ClientDisconnected?.Invoke(this, clientInfo);
            }
        }

        private async Task FillPipeAsync(NetworkStream stream, PipeWriter writer, CancellationToken cancellationToken)
        {
            const int minimumBufferSize = 4096;

            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    // 从PipeWriter获取内存
                    Memory<byte> memory = writer.GetMemory(minimumBufferSize);
                    
                    // 从网络流读取数据
                    int bytesRead = await stream.ReadAsync(memory, cancellationToken);
                    if (bytesRead == 0)
                    {
                        break; // 连接关闭
                    }

                    // 告诉PipeWriter已经写入了多少数据
                    writer.Advance(bytesRead);
                }
                catch (Exception ex)
                {
                    LogMessage($"读取数据错误: {ex.Message}");
                    break;
                }

                // 使数据对PipeReader可用
                FlushResult result = await writer.FlushAsync(cancellationToken);

                if (result.IsCompleted || result.IsCanceled)
                {
                    break;
                }
            }

            // 告诉PipeReader没有更多数据了
            await writer.CompleteAsync();
        }

        private async Task ReadPipeAsync(PipeReader reader, ClientInfo clientInfo, CancellationToken cancellationToken)
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                ReadResult result = await reader.ReadAsync(cancellationToken);
                ReadOnlySequence<byte> buffer = result.Buffer;

                // 处理接收到的数据
                while (TryReadMessage(ref buffer, out ReadOnlySequence<byte> message))
                {
                    ProcessMessage(message, clientInfo);
                }

                // 告诉Pipe我们已经处理了多少数据
                reader.AdvanceTo(buffer.Start, buffer.End);

                // 如果没有更多数据，停止读取
                if (result.IsCompleted)
                {
                    break;
                }
            }

            await reader.CompleteAsync();
        }

        private bool TryReadMessage(ref ReadOnlySequence<byte> buffer, out ReadOnlySequence<byte> message)
        {
            // 查找消息分隔符（换行符）
            SequencePosition? position = buffer.PositionOf((byte)'\n');
            
            if (position == null)
            {
                message = default;
                return false;
            }

            // 提取消息（包括换行符）
            message = buffer.Slice(0, position.Value);
            
            // 跳过处理过的数据（包括换行符）
            buffer = buffer.Slice(buffer.GetPosition(1, position.Value));
            
            return true;
        }

        private void ProcessMessage(ReadOnlySequence<byte> message, ClientInfo clientInfo)
        {
            // 将消息转换为字符串
            string text = Encoding.UTF8.GetString(message);
            
            // 移除换行符
            text = text.TrimEnd('\r', '\n');
            
            // 触发消息接收事件
            MessageReceived?.Invoke(this, new ClientMessage
            {
                Client = clientInfo,
                Message = text
            });
        }

        private void LogMessage(string message)
        {
            ServerLog?.Invoke(this, message);
        }
    }
}
